from xtquant import xtdata
import time
import pandas as pd
import numpy as np
from datetime import datetime
from typing import Dict, List, Optional

# 设置pandas选项以避免FutureWarning
pd.set_option('future.no_silent_downcasting', True)


class BlueColumnConfig:
    """蓝柱信号配置类"""
    
    # 蓝柱检测条件
    MIN_DURATION_MINUTES = 1  # 最小持续时间（分钟）
    MACD_INCREASE_RATIO = 1  # MACD值比MA值增加的比率
    
    # 缓存配置
    CACHE_TTL = 10  # 蓝柱结果缓存有效期（秒）- 分钟级数据，10秒缓存足够，减少计算阻塞


class BlueColumnCalculator:
    """蓝柱信号计算器 - 独立版本"""
    
    def __init__(self, xtdata):
        """初始化蓝柱信号计算器"""
        self.xtdata = xtdata
        self.blue_column_cache = {}  # 蓝柱结果缓存
        self.last_blue_check = {}  # 上次蓝柱检查时间
    
    def calculate_macd_ma(self, macd_histogram: pd.Series, period: int = 5) -> pd.Series:
        """
        计算MACD柱状图的移动平均线

        Args:
            macd_histogram: MACD柱状图数据
            period: 移动平均周期，默认5

        Returns:
            MACD柱状图的移动平均线
        """
        return macd_histogram.rolling(window=period, min_periods=1).mean()

    def detect_blue_column_signals(self, macd_histogram: pd.Series, 
                                   macd_ma: pd.Series,
                                   df_minutes: pd.DataFrame,
                                   price_col: str = 'close') -> List[Dict[str, any]]:
        """
        检测蓝柱信号：模拟同花顺的蓝柱显示逻辑

        同花顺蓝柱特点：
        1. 只在MACD柱状图为负值时显示蓝柱（绿色柱状图向上收敛）
        2. 需要MACD柱状图 > MA(MACD柱状图, 5)
        3. 蓝柱通常在死叉范围内（DIFF < DEA）
        4. 表示MACD柱状图开始向上收敛，可能预示反弹

        Args:
            macd_histogram: MACD柱状图数据
            macd_ma: MACD柱状图的5周期移动平均线
            df_minutes: 分钟数据DataFrame
            price_col: 价格列名称，默认'close'

        Returns:
            蓝柱信号列表
        """
        blue_signals = []

        # 检测蓝柱条件：MACD柱状图 > MA(MACD柱状图, 5) 且 MACD柱状图 < 0
        blue_condition = (macd_histogram > macd_ma) & (macd_histogram < 0)

        # 找到蓝柱状态的转折点
        blue_condition_shifted = blue_condition.shift(1).fillna(False)

        # 蓝柱开始：从False变为True
        blue_start_mask = (~blue_condition_shifted) & blue_condition
        # 蓝柱结束：从True变为False
        blue_end_mask = blue_condition_shifted & (~blue_condition)

        # 使用位置索引而不是标签索引
        blue_start_indices = blue_start_mask[blue_start_mask].index.tolist()
        blue_end_indices = blue_end_mask[blue_end_mask].index.tolist()

        # 记录蓝柱开始信号
        for start_idx in blue_start_indices:
            try:
                # 获取当前时间的价格
                if start_idx not in df_minutes.index:
                    # 如果不在索引中，使用位置索引
                    if start_idx < len(df_minutes):
                        t = df_minutes.iloc[start_idx]['time']
                        price_at_time = df_minutes.iloc[start_idx].get(price_col, 0)
                    else:
                        continue
                else:
                    t = df_minutes.loc[start_idx, 'time']
                    price_at_time = df_minutes.loc[start_idx].get(price_col, 0)

                if pd.isna(t):
                    continue

                time_str = pd.to_datetime(t).strftime('%Y-%m-%d %H:%M:%S')

                # 获取对应的MACD值
                if start_idx < len(macd_histogram):
                    macd_value = macd_histogram.iloc[start_idx]
                    macd_ma_value = macd_ma.iloc[start_idx]
                else:
                    macd_value = 0
                    macd_ma_value = 0

                # 计算蓝柱持续时间（找到对应的结束时间）
                duration_minutes = 0
                end_time_str = None

                # 查找对应的结束时间
                for end_idx in blue_end_indices:
                    if end_idx > start_idx:
                        duration_minutes = end_idx - start_idx
                        if end_idx < len(df_minutes):
                            end_t = df_minutes.iloc[end_idx]['time']
                            end_time_str = pd.to_datetime(end_t).strftime('%Y-%m-%d %H:%M:%S')
                        break

                # 只记录持续时间>=1分钟且MACD值足够大的蓝柱
                min_duration = BlueColumnConfig.MIN_DURATION_MINUTES
                min_ratio = BlueColumnConfig.MACD_INCREASE_RATIO
                
                if duration_minutes >= min_duration and macd_value > macd_ma_value * min_ratio:
                    blue_signals.append({
                        'time': time_str,
                        'price': float(price_at_time),
                        'macd_value': float(macd_value),
                        'macd_ma_value': float(macd_ma_value),
                        'signal_type': 'blue_column_start',
                        'duration_minutes': duration_minutes,
                        'end_time': end_time_str
                    })
            except Exception:
                # 跳过错误，继续处理下一个
                continue

        return blue_signals

    def get_minute_data(self, stock_code: str, force_download: bool = False) -> Optional[pd.DataFrame]:
        """
        获取分钟级数据
        优化：先尝试本地数据，避免每次都重新下载（提高性能）
        
        Args:
            stock_code: 股票代码
            force_download: 是否强制重新下载，默认False
        """
        today = datetime.now().strftime("%Y%m%d")
        
        # 先尝试获取本地数据（避免不必要的下载）
        if not force_download:
            try:
                data = self.xtdata.get_local_data(
                    stock_list=[stock_code], 
                    period='1m',
                    start_time=today, 
                    end_time=today
                )
                
                if data and stock_code in data:
                    df = data[stock_code]
                    if not df.empty:
                        # 找到时间列（可能是'time'或包含'time'的列）
                        time_col = next((col for col in df.columns if 'time' in col.lower()), None)
                        
                        if time_col:
                            try:
                                # 获取最后一条数据的时间
                                last_time_raw = df[time_col].iloc[-1]
                                last_time = pd.to_datetime(last_time_raw)
                                if isinstance(last_time, pd.Timestamp):
                                    last_time_py = last_time.to_pydatetime()
                                    time_diff = (datetime.now() - last_time_py).total_seconds()
                                    
                                    # 如果数据足够新（40秒内），直接使用
                                    if time_diff < 40:
                                        return df
                                    # 否则触发重新下载（数据可能已更新）
                            except:
                                pass
            except Exception:
                pass
        
        # 本地数据不存在或过期，重新下载
        try:
            # 强制下载最新数据
            self.xtdata.download_history_data(
                stock_code=stock_code, 
                period="1m", 
                start_time=today, 
                end_time=today
            )
            time.sleep(0.1)  # 缩短等待时间，提高响应速度
            
            # 获取本地分钟数据
            data = self.xtdata.get_local_data(
                stock_list=[stock_code], 
                period='1m',
                start_time=today, 
                end_time=today
            )
            
            if not data or stock_code not in data:
                return None
            
            df = data[stock_code]
            if df.empty:
                return None
            
            return df
        except Exception:
            return None
    
    def calculate_macd_for_blue_column(self, prices: pd.Series, fast: int = 12, 
                                       slow: int = 26, signal: int = 9) -> tuple:
        """
        计算MACD用于蓝柱信号检测
        使用同花顺方法：简单初始化
        """
        if not isinstance(prices, pd.Series):
            prices = pd.Series(prices)
        
        series = prices.astype(float).reset_index(drop=True)
        
        def calc_ema_mode_a(data, period):
            """同花顺方法：第一个EMA = 第一个收盘价"""
            alpha = 2 / (period + 1)
            ema = np.zeros(len(data))
            ema[0] = data[0]  # 第一个EMA = 第一个收盘价
            for i in range(1, len(data)):
                ema[i] = alpha * data[i] + (1 - alpha) * ema[i - 1]
            return ema
        
        # 计算快慢线
        ema_fast = calc_ema_mode_a(series.values, fast)
        ema_slow = calc_ema_mode_a(series.values, slow)
        
        # 计算DIFF
        diff = ema_fast - ema_slow
        
        # 计算DEA（对DIFF做EMA）
        dea = calc_ema_mode_a(diff, signal)
        
        # 转为Series
        diff_series = pd.Series(diff, index=series.index)
        dea_series = pd.Series(dea, index=series.index)
        macd_histogram = 2 * (diff_series - dea_series)
        
        return diff_series, dea_series, macd_histogram
    
    def detect_blue_signals(self, stock_code: str, 
                            fast: int = 12, slow: int = 26, signal: int = 9,
                            include_auction: bool = True, 
                            return_all_signals: bool = False,
                            macd_cache_data: Optional[Dict] = None) -> Dict[str, any]:
        """
        检测蓝柱信号并返回结果

        Args:
            stock_code: 股票代码
            fast: 快线EMA周期，默认12
            slow: 慢线EMA周期，默认26
            signal: 信号线EMA周期，默认9
            include_auction: 是否包含集合竞价数据（9:15-9:25），默认True
            return_all_signals: 是否返回所有信号，默认False（只返回最近5分钟内）
            macd_cache_data: MACD缓存数据（可选），如果提供则直接使用，避免重复计算MACD

        Returns:
            蓝柱信号列表
        """
        current_time = time.time()
        
        # 检查缓存（10秒有效期，分钟级数据足够）
        if stock_code in self.blue_column_cache:
            cached_time_raw = self.last_blue_check.get(stock_code, 0)
            # 兼容可能为字符串的时间值
            if isinstance(cached_time_raw, (int, float)):
                cached_time = float(cached_time_raw)
            elif isinstance(cached_time_raw, str):
                try:
                    # 尝试直接转为float时间戳
                    cached_time = float(cached_time_raw)
                except Exception:
                    try:
                        # 尝试解析常见的时间字符串格式
                        parsed_dt = pd.to_datetime(cached_time_raw, errors='coerce')
                        cached_time = parsed_dt.to_pydatetime().timestamp() if pd.notna(parsed_dt) else 0.0
                    except Exception:
                        cached_time = 0.0
            else:
                cached_time = 0.0

            if (current_time - cached_time) < BlueColumnConfig.CACHE_TTL:
                # 在缓存有效期内，直接返回缓存结果
                return self.blue_column_cache[stock_code]
        
        # 如果提供了MACD缓存数据，尝试直接使用
        if macd_cache_data and isinstance(macd_cache_data, dict):
            macd_histogram_dict = macd_cache_data.get('macd_histogram_series', {})
            macd_series_data = macd_cache_data.get('macd_series', {})
            df_minutes_data = macd_cache_data.get('df_minutes', [])
            
            # 检查MACD缓存数据是否完整
            if macd_histogram_dict and macd_series_data:
                # 使用MACD缓存数据计算蓝柱
                try:
                    # 需要获取对应的分钟数据以获取时间信息（如果缓存中没有）
                    if not df_minutes_data:
                        df = self.get_minute_data(stock_code, force_download=False)
                        if df is None or df.empty:
                            # 如果没有分钟数据，降级为重新计算
                            pass
                        else:
                            # 使用MACD缓存的histogram计算蓝柱
                            # histogram_dict的key是时间字符串，需要转换为Series并与df对齐
                            return self._calculate_blue_from_macd_cache(
                                stock_code, macd_histogram_dict, macd_series_data, df, 
                                current_time, return_all_signals
                            )
                    else:
                        # 缓存中有分钟数据，直接使用
                        df = pd.DataFrame(df_minutes_data)
                        return self._calculate_blue_from_macd_cache(
                            stock_code, macd_histogram_dict, macd_series_data, df, 
                            current_time, return_all_signals
                        )
                except Exception as e:
                    import logging
                    logging.warning(f"{stock_code} 使用MACD缓存计算蓝柱失败，降级为重新计算: {e}")
                    import traceback
                    logging.debug(f"异常详情: {traceback.format_exc()}")
        
        # 缓存过期或不存在，需要重新计算
        # 不强制下载，使用智能缓存逻辑（先尝试本地数据，如果足够新就不重新下载）
        force_download = False
        
        # 获取分钟数据
        df = self.get_minute_data(stock_code, force_download=force_download)
        if df is None or df.empty:
            result = {'blue_column': [], 'latest_macd': {}}
            self.blue_column_cache[stock_code] = result
            self.last_blue_check[stock_code] = current_time
            return result

        # 确定价格列和时间列
        price_col = next((col for col in ['close', 'lastPrice', 'price', 'last', 'current'] 
                         if col in df.columns), None)
        time_col = next((col for col in df.columns if 'time' in col.lower()), None)
        
        if not price_col or not time_col:
            result = {'blue_column': [], 'latest_macd': {}}
            self.blue_column_cache[stock_code] = result
            self.last_blue_check[stock_code] = current_time
            return result

        # 转换时间戳
        try:
            if df[time_col].dtype in ['int64', 'float64']:
                df['time'] = pd.to_datetime(df[time_col], unit='ms') + pd.Timedelta(hours=8)
            else:
                df['time'] = pd.to_datetime(df[time_col], errors='coerce')

            df = df[df['time'].notna()].sort_values('time')
            if df.empty:
                result = {'blue_column': [], 'latest_macd': {}}
                self.blue_column_cache[stock_code] = result
                self.last_blue_check[stock_code] = current_time
                return result
        except Exception:
            result = {'blue_column': [], 'latest_macd': {}}
            self.blue_column_cache[stock_code] = result
            self.last_blue_check[stock_code] = current_time
            return result

        # 过滤无效数据和午休时间
        df_filtered = df[df[price_col] > 0].copy()
        df_filtered['time_only'] = df_filtered['time'].dt.time

        if include_auction:
            # 包含集合竞价：保留 9:15-11:30 和 13:00-15:00
            morning_start = pd.Timestamp('09:15:00').time()
            morning_end = pd.Timestamp('11:30:00').time()
            afternoon_start = pd.Timestamp('13:00:00').time()
            afternoon_end = pd.Timestamp('15:00:00').time()

            df_filtered = df_filtered[
                ((df_filtered['time_only'] >= morning_start) & (df_filtered['time_only'] <= morning_end)) |
                ((df_filtered['time_only'] >= afternoon_start) & (df_filtered['time_only'] <= afternoon_end))
                ]
        else:
            # 不包含集合竞价：保留 9:30-11:30 和 13:00-15:00
            morning_start = pd.Timestamp('09:30:00').time()
            morning_end = pd.Timestamp('11:30:00').time()
            afternoon_start = pd.Timestamp('13:00:00').time()
            afternoon_end = pd.Timestamp('15:00:00').time()

            df_filtered = df_filtered[
                ((df_filtered['time_only'] >= morning_start) & (df_filtered['time_only'] <= morning_end)) |
                ((df_filtered['time_only'] >= afternoon_start) & (df_filtered['time_only'] <= afternoon_end))
                ]

        if df_filtered.empty:
            result = {'blue_column': [], 'latest_macd': {}}
            self.blue_column_cache[stock_code] = result
            self.last_blue_check[stock_code] = current_time
            return result
        
        df_minutes = df_filtered.copy()
        
        # 计算MACD - 使用close列，如果没有则使用价格列
        price_series = df_minutes.get('close', df_minutes.get(price_col))
        if price_series is None or price_series.empty:
            result = {'blue_column': [], 'latest_macd': {}}
            self.blue_column_cache[stock_code] = result
            self.last_blue_check[stock_code] = current_time
            return result
        
        # 计算MACD
        diff, dea, histogram = self.calculate_macd_for_blue_column(
                price_series,
                fast=fast,
                slow=slow,
                signal=signal
            )

        # 计算MACD柱状图的移动平均线
        macd_ma = self.calculate_macd_ma(histogram, period=5)

        # 检测蓝柱信号
        all_blue_signals = self.detect_blue_column_signals(histogram, macd_ma, df_minutes, price_col)
        
        # 根据参数决定返回所有信号还是只返回最近5分钟内的信号
        current_datetime = datetime.now()
        blue_column_signals = []
        for signal in all_blue_signals:
            signal_time_str = signal.get('time', '')
            if signal_time_str:
                try:
                    signal_time = datetime.strptime(signal_time_str, '%Y-%m-%d %H:%M:%S')
                    time_diff = (current_datetime - signal_time).total_seconds()
                    # 如果 return_all_signals=True，返回所有信号；否则只返回5分钟内的
                    if return_all_signals or time_diff <= 300:
                        blue_column_signals.append(signal)
                except:
                    pass
        
        # 调试：输出检测到的信号数量和最新K线时间
        if blue_column_signals:
            import logging
            # 获取数据的最新时间
            if not df_minutes.empty and 'time' in df_minutes.columns:
                latest_time = df_minutes['time'].max()
                time_str = pd.to_datetime(latest_time).strftime('%H:%M:%S')
                current_time = datetime.now().strftime('%H:%M:%S')
                logging.info(f"🔵 {stock_code} 最新K线: {time_str}, 当前时间: {current_time}, 蓝柱信号数: {len(blue_column_signals)}")
            
            if return_all_signals:
                logging.info(f"🔵 {stock_code} 检测到 {len(blue_column_signals)} 个蓝柱信号（全部）")
            else:
                logging.info(f"🔵 {stock_code} 检测到 {len(blue_column_signals)} 个最近的蓝柱信号（最近5分钟内）")

        # 构建最新MACD信息
        latest_macd = {}
        try:
            if not diff.empty and not dea.empty:
                valid_mask = diff.notna() & dea.notna()
                if valid_mask.any():
                    last_idx = diff[valid_mask].index[-1]
                    latest_diff = diff.loc[last_idx]
                    latest_dea = dea.loc[last_idx]
                    if not histogram.empty and last_idx in histogram.index and not pd.isna(histogram.loc[last_idx]):
                        latest_histogram = histogram.loc[last_idx]
                    else:
                        latest_histogram = 2 * (latest_diff - latest_dea)

                    latest_macd = {
                        'diff': float(latest_diff),
                        'dea': float(latest_dea),
                        'macd': float(latest_histogram),
                        'trend': 'bullish' if latest_diff > latest_dea else 'bearish'
                    }
        except Exception:
            latest_macd = {}

        result = {
            'blue_column': blue_column_signals,
            'latest_macd': latest_macd
        }
        
        # 缓存结果
        self.blue_column_cache[stock_code] = result
        self.last_blue_check[stock_code] = current_time
        
        return result
    
    def _calculate_blue_from_macd_cache(self, stock_code: str, macd_histogram_dict: Dict, 
                                      macd_series_data: Dict, df_minutes: pd.DataFrame,
                                      current_time: float, return_all_signals: bool) -> Dict[str, any]:
        """
        使用MACD缓存数据计算蓝柱信号（优化版本，避免重复计算MACD）
        
        Args:
            stock_code: 股票代码
            macd_histogram_dict: MACD柱状图字典（key为时间字符串，value为MACD值）
            macd_series_data: MACD序列数据字典（key为时间字符串）
            df_minutes: 分钟数据DataFrame（用于获取时间信息）
            current_time: 当前时间戳
            return_all_signals: 是否返回所有信号
            
        Returns:
            蓝柱信号结果字典
        """
        try:
            # 确保df_minutes有time列
            if 'time' not in df_minutes.columns:
                # 尝试找到时间列
                time_col = next((col for col in df_minutes.columns if 'time' in col.lower()), None)
                if time_col:
                    try:
                        if df_minutes[time_col].dtype in ['int64', 'float64']:
                            df_minutes['time'] = pd.to_datetime(df_minutes[time_col], unit='ms') + pd.Timedelta(hours=8)
                        else:
                            df_minutes['time'] = pd.to_datetime(df_minutes[time_col], errors='coerce')
                    except:
                        pass
            
            # 确定价格列
            price_col = next((col for col in ['close', 'lastPrice', 'price', 'last', 'current'] 
                             if col in df_minutes.columns), None)
            
            if 'time' not in df_minutes.columns or not price_col:
                result = {'blue_column': [], 'latest_macd': {}}
                self.blue_column_cache[stock_code] = result
                self.last_blue_check[stock_code] = current_time
                return result
            
            # 确保time列是datetime类型
            if not pd.api.types.is_datetime64_any_dtype(df_minutes['time']):
                df_minutes['time'] = pd.to_datetime(df_minutes['time'], errors='coerce')
            
            df_minutes = df_minutes[df_minutes['time'].notna()].sort_values('time')
            if df_minutes.empty:
                result = {'blue_column': [], 'latest_macd': {}}
                self.blue_column_cache[stock_code] = result
                self.last_blue_check[stock_code] = current_time
                return result
            
            # 将macd_histogram_dict与df_minutes对齐，转换为Series
            # 需要确保histogram的索引与df_minutes对齐
            histogram_values = []
            for idx, row in df_minutes.iterrows():
                time_str = pd.to_datetime(row['time']).strftime('%Y-%m-%d %H:%M:%S')
                histogram_val = macd_histogram_dict.get(time_str, 0)
                histogram_values.append(histogram_val)
            
            macd_histogram_series = pd.Series(histogram_values, index=df_minutes.index)
            
            # 计算MACD柱状图的移动平均线
            macd_ma = self.calculate_macd_ma(macd_histogram_series, period=5)
            
            # 检测蓝柱信号
            all_blue_signals = self.detect_blue_column_signals(macd_histogram_series, macd_ma, df_minutes, price_col)
            
            # 根据参数决定返回所有信号还是只返回最近5分钟内的信号
            current_datetime = datetime.now()
            blue_column_signals = []
            for signal in all_blue_signals:
                signal_time_str = signal.get('time', '')
                if signal_time_str:
                    try:
                        signal_time = datetime.strptime(signal_time_str, '%Y-%m-%d %H:%M:%S')
                        time_diff = (current_datetime - signal_time).total_seconds()
                        # 如果 return_all_signals=True，返回所有信号；否则只返回5分钟内的
                        if return_all_signals or time_diff <= 300:
                            blue_column_signals.append(signal)
                    except:
                        pass
            
            # 构建最新MACD信息（从macd_series_data获取）
            latest_macd = {}
            try:
                if macd_series_data:
                    # 获取最新的MACD数据（按时间排序）
                    latest_time = max(macd_series_data.keys())
                    latest_data = macd_series_data[latest_time]
                    latest_macd = {
                        'diff': latest_data.get('diff', 0),
                        'dea': latest_data.get('dea', 0),
                        'macd': latest_data.get('macd', 0),
                        'trend': 'bullish' if latest_data.get('diff', 0) > latest_data.get('dea', 0) else 'bearish'
                    }
            except Exception:
                latest_macd = {}
            
            result = {
                'blue_column': blue_column_signals,
                'latest_macd': latest_macd
            }
            
            # 缓存结果
            self.blue_column_cache[stock_code] = result
            self.last_blue_check[stock_code] = current_time
            
            return result
        except Exception as e:
            import logging
            logging.error(f"{stock_code} 使用MACD缓存计算蓝柱异常: {e}")
            # 降级为返回空结果
            result = {'blue_column': [], 'latest_macd': {}}
            self.blue_column_cache[stock_code] = result
            self.last_blue_check[stock_code] = current_time
            return result
    
    def clear_cache(self, stock_code: str = None):
        """清除缓存"""
        if stock_code:
            if stock_code in self.blue_column_cache:
                del self.blue_column_cache[stock_code]
            if stock_code in self.last_blue_check:
                del self.last_blue_check[stock_code]
        else:
                self.blue_column_cache.clear()
                self.last_blue_check.clear()

    def add_realtime_signal(self, stock_code: str, signal: Dict):
        """将实时检测到的蓝柱信号写入缓存，便于状态查询/后续判断保持一致。"""
        try:
            if not isinstance(signal, dict):
                return
            cached = self.blue_column_cache.get(stock_code, {'blue_column': [], 'latest_macd': {}})
            blue_list = cached.get('blue_column', [])
            blue_list.append(signal)
            # 仅保留最近的若干条，避免无限增长
            if len(blue_list) > 50:
                blue_list = blue_list[-50:]
            cached['blue_column'] = blue_list
            self.blue_column_cache[stock_code] = cached
            self.last_blue_check[stock_code] = time.time()
        except Exception:
            pass
    
    def get_cache_stats(self) -> Dict:
        """获取缓存统计信息"""
        return {
            'cache_size': len(self.blue_column_cache),
            'cached_stocks': list(self.blue_column_cache.keys())
        }
    
    def get_latest_blue_signal(self, stock_code: str) -> Optional[Dict]:
        """获取指定股票的最新蓝柱信号"""
        result = self.detect_blue_signals(stock_code, include_auction=True)
        blue_signals = result.get('blue_column', [])
        if blue_signals and len(blue_signals) > 0:
            return blue_signals[-1]  # 返回最新的蓝柱信号
        return None
    
    def check_realtime_blue_column(self, stock_code: str, current_price: float) -> Optional[Dict]:
        """
        实时检测蓝柱状态（轻量级，只更新最后一根K线）
        用于实时监控，不重新计算全部MACD
        
        Args:
            stock_code: 股票代码
            current_price: 当前价格（从tick获取）
            
        Returns:
            如果卖出蓝柱信号，返回信号字典；否则返回None
        """
        try:
            # 获取分钟数据
            df = self.get_minute_data(stock_code, force_download=False)
            if df is None or df.empty:
                return None
            
            # 确定价格列和时间列
            price_col = next((col for col in ['close', 'lastPrice', 'price', 'last', 'current'] 
                         if col in df.columns), None)
            time_col = next((col for col in df.columns if 'time' in col.lower()), None)
            
            if not price_col or not time_col:
                return None
            
            # 更新最后一根K线的收盘价为当前价格
            if len(df) > 0:
                df.loc[df.index[-1], price_col] = current_price
                
                # 只重新计算MACD的最后几个值（EMA只需要最后26个值）
                price_series = df[price_col].astype(float)
                diff, dea, histogram = self.calculate_macd_for_blue_column(price_series)
                
                # 计算MACD柱状图的移动平均线
                macd_ma = self.calculate_macd_ma(histogram, period=5)
                
                # 获取最后一根K线的值
                last_idx = len(histogram) - 1
                last_macd = histogram.iloc[last_idx]
                last_ma = macd_ma.iloc[last_idx]
                
                # 检查是否满足蓝柱条件
                if last_macd < 0 and last_macd > last_ma:
                    # 满足蓝柱条件，返回实时信号
                    last_time = df[time_col].iloc[last_idx]
                    # 规范化时间：支持毫秒时间戳/字符串，并加8小时
                    try:
                        if pd.api.types.is_numeric_dtype(df[time_col]):
                            # 认为为毫秒时间戳
                            ts = pd.to_datetime(last_time, unit='ms') + pd.Timedelta(hours=8)
                        else:
                            ts = pd.to_datetime(last_time, errors='coerce')
                            if pd.isna(ts):
                                ts = pd.Timestamp.now()
                    except Exception:
                        ts = pd.Timestamp.now()
                    # 为实时信号使用当前时间，避免因K线时间在未来一分钟边界导致负延迟
                    time_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                    
                    return {
                        'time': time_str,
                        'price': float(current_price),
                        'macd_value': float(last_macd),
                        'macd_ma_value': float(last_ma),
                        'signal_type': 'blue_column_realtime',
                        'is_realtime': True
                    }
            
            return None
        except Exception as e:
            import logging
            logging.error(f"实时蓝柱检测异常: {e}")
            return None
    
    def get_blue_column_status_str(self, stock_code: str) -> str:
        """获取蓝柱状态的字符串描述，用于日志输出"""
        # 1) 优先获取全部蓝柱信号（不过滤5分钟窗口），以便展示“最新一次”的距离
        result = self.detect_blue_signals(stock_code, include_auction=True, return_all_signals=True)
        blue_signals = result.get('blue_column', [])

        # 2) 若为空，尝试从缓存中回退
        if (not blue_signals) and stock_code in self.blue_column_cache:
            cache_item = self.blue_column_cache.get(stock_code) or {}
            blue_signals = cache_item.get('blue_column', []) or []

        if blue_signals:
            latest = blue_signals[-1]
            signal_time = latest.get('time', '')
            signal_price = latest.get('price', 0)
            macd_value = latest.get('macd_value', 0)
            macd_ma_value = latest.get('macd_ma_value', 0)

            # 计算时间差
            try:
                signal_datetime = datetime.strptime(signal_time, '%Y-%m-%d %H:%M:%S')
                time_diff = (datetime.now() - signal_datetime).total_seconds()
                return f"蓝柱信号: {signal_time}, 价格: {signal_price:.3f}, MACD: {macd_value:.6f}, MA: {macd_ma_value:.6f}, 距现在: {time_diff:.1f}秒"
            except Exception:
                return f"蓝柱信号: {signal_time}, 价格: {signal_price:.3f}, MACD: {macd_value:.6f}, MA: {macd_ma_value:.6f}"
        return "暂无蓝柱信号"


# 测试代码
if __name__ == "__main__":
    calculator = BlueColumnCalculator(xtdata)
    
    stock_code = "513090.SH"
    print("=" * 80)
    print("蓝柱信号独立计算器测试")
    print("=" * 80)

    # 检测蓝柱信号
    result = calculator.detect_blue_signals(stock_code, include_auction=True)
    
    print(f"\n股票代码: {stock_code}")
    print(f"蓝柱信号数量: {len(result['blue_column'])}")
    
    if result['blue_column']:
        print("\n【蓝柱信号】")
        for i, signal in enumerate(result['blue_column'], 1):
            print(f"  {i}. 时间: {signal['time']}, 价格: {signal['price']:.4f}")
            print(f"     MACD: {signal['macd_value']:.6f}, MA: {signal['macd_ma_value']:.6f}")
            print(f"     持续时间: {signal['duration_minutes']}分钟")
    
    if result['latest_macd']:
        print(f"\n【最新MACD值】")
        print(f"  DIFF: {result['latest_macd']['diff']:.6f}")
        print(f"  DEA: {result['latest_macd']['dea']:.6f}")
        print(f"  MACD: {result['latest_macd']['macd']:.6f}")
        print(f"  趋势: {'多头' if result['latest_macd']['trend'] == 'bullish' else '空头'}")
    
    print("=" * 80)
    print("缓存统计:", calculator.get_cache_stats())
    print("=" * 80)
